/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.broker;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.broker.client.ClientHousekeepingService;
import com.alibaba.rocketmq.broker.client.ConsumerIdsChangeListener;
import com.alibaba.rocketmq.broker.client.ConsumerManager;
import com.alibaba.rocketmq.broker.client.DefaultConsumerIdsChangeListener;
import com.alibaba.rocketmq.broker.client.ProducerManager;
import com.alibaba.rocketmq.broker.client.net.Broker2Client;
import com.alibaba.rocketmq.broker.client.rebalance.RebalanceLockManager;
import com.alibaba.rocketmq.broker.filtersrv.FilterServerManager;
import com.alibaba.rocketmq.broker.longpolling.PullRequestHoldService;
import com.alibaba.rocketmq.broker.mqtrace.ConsumeMessageHook;
import com.alibaba.rocketmq.broker.mqtrace.SendMessageHook;
import com.alibaba.rocketmq.broker.offset.ConsumerOffsetManager;
import com.alibaba.rocketmq.broker.out.BrokerOuterAPI;
import com.alibaba.rocketmq.broker.processor.AdminBrokerProcessor;
import com.alibaba.rocketmq.broker.processor.ClientManageProcessor;
import com.alibaba.rocketmq.broker.processor.EndTransactionProcessor;
import com.alibaba.rocketmq.broker.processor.PullMessageProcessor;
import com.alibaba.rocketmq.broker.processor.QueryMessageProcessor;
import com.alibaba.rocketmq.broker.processor.SendMessageProcessor;
import com.alibaba.rocketmq.broker.slave.SlaveSynchronize;
import com.alibaba.rocketmq.broker.subscription.SubscriptionGroupManager;
import com.alibaba.rocketmq.broker.topic.TopicConfigManager;
import com.alibaba.rocketmq.common.BrokerConfig;
import com.alibaba.rocketmq.common.DataVersion;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.ThreadFactoryImpl;
import com.alibaba.rocketmq.common.TopicConfig;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.constant.LoggerName;
import com.alibaba.rocketmq.common.constant.PermName;
import com.alibaba.rocketmq.common.namesrv.RegisterBrokerResult;
import com.alibaba.rocketmq.common.protocol.RequestCode;
import com.alibaba.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
import com.alibaba.rocketmq.remoting.RPCHook;
import com.alibaba.rocketmq.remoting.RemotingServer;
import com.alibaba.rocketmq.remoting.netty.NettyClientConfig;
import com.alibaba.rocketmq.remoting.netty.NettyRemotingServer;
import com.alibaba.rocketmq.remoting.netty.NettyRequestProcessor;
import com.alibaba.rocketmq.remoting.netty.NettyServerConfig;
import com.alibaba.rocketmq.store.DefaultMessageStore;
import com.alibaba.rocketmq.store.MessageStore;
import com.alibaba.rocketmq.store.config.BrokerRole;
import com.alibaba.rocketmq.store.config.MessageStoreConfig;
import com.alibaba.rocketmq.store.stats.BrokerStats;
import com.alibaba.rocketmq.store.stats.BrokerStatsManager;

import java.net.InetSocketAddress;

/**
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-26
 */
public class BrokerController {
	private static final Logger log = LoggerFactory.getLogger(LoggerName.BrokerLoggerName);
	private final BrokerConfig brokerConfig;
	private final NettyServerConfig nettyServerConfig;
	private final NettyClientConfig nettyClientConfig;
	private final MessageStoreConfig messageStoreConfig;
	private final DataVersion configDataVersion = new DataVersion();
	private final ConsumerOffsetManager consumerOffsetManager;
	private final ConsumerManager consumerManager;
	private final ProducerManager producerManager;
	private final ClientHousekeepingService clientHousekeepingService;
	private final PullMessageProcessor pullMessageProcessor;
	private final PullRequestHoldService pullRequestHoldService;
	private final Broker2Client broker2Client;
	private final SubscriptionGroupManager subscriptionGroupManager;
	private final ConsumerIdsChangeListener consumerIdsChangeListener;
	private final RebalanceLockManager rebalanceLockManager = new RebalanceLockManager();
	private final BrokerOuterAPI brokerOuterAPI;
	private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("BrokerControllerScheduledThread"));
	private final SlaveSynchronize slaveSynchronize;
	private MessageStore messageStore;
	private RemotingServer remotingServer;
	private TopicConfigManager topicConfigManager;
	private ExecutorService sendMessageExecutor;
	private ExecutorService pullMessageExecutor;
	private ExecutorService adminBrokerExecutor;
	private ExecutorService clientManageExecutor;
	private boolean updateMasterHAServerAddrPeriodically = false;

	private BrokerStats brokerStats;
	private final BlockingQueue<Runnable> sendThreadPoolQueue;

	private final BlockingQueue<Runnable> pullThreadPoolQueue;

	private final FilterServerManager filterServerManager;

	private final BrokerStatsManager brokerStatsManager;
	private InetSocketAddress storeHost;

	public BrokerController(//
			final BrokerConfig brokerConfig, //
			final NettyServerConfig nettyServerConfig, //
			final NettyClientConfig nettyClientConfig, //
			final MessageStoreConfig messageStoreConfig //
	) {
		this.brokerConfig = brokerConfig;
		this.nettyServerConfig = nettyServerConfig;
		this.nettyClientConfig = nettyClientConfig;
		this.messageStoreConfig = messageStoreConfig;
		this.consumerOffsetManager = new ConsumerOffsetManager(this);
		/**
		 * 默认初始化了"SELF_TEST_TOPIC"、"TBW102"（自动创建Topic功能是否开启BrokerConfig.autoCreateTopicEnable=false，线上建议关闭）
		 * 、"BenchmarkTest"、BrokerClusterName（集群名称）、BrokerName（Broker的名称）、"OFFSET_MOVED_EVENT"
		 * 这5个topic的信息并存入了topicConfigTable变量中，在向NameServer注册时发给NameServer进行登记；
		 */
		this.topicConfigManager = new TopicConfigManager(this);
		this.pullMessageProcessor = new PullMessageProcessor(this);
		this.pullRequestHoldService = new PullRequestHoldService(this);
		this.consumerIdsChangeListener = new DefaultConsumerIdsChangeListener(this);
		this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener);
		this.producerManager = new ProducerManager();
		this.clientHousekeepingService = new ClientHousekeepingService(this);
		this.broker2Client = new Broker2Client(this);
		this.subscriptionGroupManager = new SubscriptionGroupManager(this);
		this.brokerOuterAPI = new BrokerOuterAPI(nettyClientConfig);
		this.filterServerManager = new FilterServerManager(this);

		if (this.brokerConfig.getNamesrvAddr() != null) {
			this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
			log.info("user specfied name server address: {}", this.brokerConfig.getNamesrvAddr());
		}

		this.slaveSynchronize = new SlaveSynchronize(this);

		this.sendThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getSendThreadPoolQueueCapacity());

		this.pullThreadPoolQueue = new LinkedBlockingQueue<Runnable>(this.brokerConfig.getPullThreadPoolQueueCapacity());

		this.brokerStatsManager = new BrokerStatsManager(this.brokerConfig.getBrokerClusterName());
		this.setStoreHost(new InetSocketAddress(this.getBrokerConfig().getBrokerIP1(), this.getNettyServerConfig().getListenPort()));
	}

	public boolean initialize() {
		boolean result = true;

		// 加载topics.json文件，分别将文件的数据存入TopicConfigManager对象中
		result = result && this.topicConfigManager.load();
		// /加载consumerOffset.json文件，分别将文件的数据存入ConsumerOffsetManager对象中
		result = result && this.consumerOffsetManager.load();
		// 加载subscriptionGroup.json文件，分别将文件的数据存入SubscriptionGroupManager对象中
		result = result && this.subscriptionGroupManager.load();

		if (result) {
			try {
				// 该对象是应用层访问存储层的访问类
				// 其中启动了HAService服务======用于消息数据同步
				this.messageStore = new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager);
			} catch (IOException e) {
				result = false;
				e.printStackTrace();
			}
		}

		result = result && this.messageStore.load();

		if (result) {
			// 初始化Netty服务端NettyRemotingServer对象
			this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.clientHousekeepingService);

			// 初始化发送消息线程池
			this.sendMessageExecutor = new ThreadPoolExecutor(//
					this.brokerConfig.getSendMessageThreadPoolNums(), //
					this.brokerConfig.getSendMessageThreadPoolNums(), //
					1000 * 60, //
					TimeUnit.MILLISECONDS, //
					this.sendThreadPoolQueue, //
					new ThreadFactoryImpl("SendMessageThread_"));
			// 初始化拉取消息线程池
			this.pullMessageExecutor = new ThreadPoolExecutor(//
					this.brokerConfig.getPullMessageThreadPoolNums(), //
					this.brokerConfig.getPullMessageThreadPoolNums(), //
					1000 * 60, //
					TimeUnit.MILLISECONDS, //
					this.pullThreadPoolQueue, //
					new ThreadFactoryImpl("PullMessageThread_"));

			// 管理Broker线程池
			this.adminBrokerExecutor = Executors.newFixedThreadPool(this.brokerConfig.getAdminBrokerThreadPoolNums(), new ThreadFactoryImpl("AdminBrokerThread_"));
			// 客户端管理线程池
			this.clientManageExecutor = Executors.newFixedThreadPool(this.brokerConfig.getClientManageThreadPoolNums(), new ThreadFactoryImpl("ClientManageThread_"));
			/**
			 * 注册事件处理器，
			 * 包括发送消息事件处理器（SendMessageProcessor）、
			 * 拉取消息事件处理器、
			 * 查询消息事件处理器（QueryMessageProcessor，包括客户端的心跳事件、注销事件、获取消费者列表事件、更新更新和查询消费进度consumerOffset）、
			 * 客户端管理事件处理器、结束事务处理器（EndTransactionProcessor）、
			 * 默认事件处理器（AdminBrokerProcessor）。
			 */
			this.registerProcessor();

			this.brokerStats = new BrokerStats((DefaultMessageStore) this.messageStore);

			// TODO remove in future
			final long initialDelay = UtilAll.computNextMorningTimeMillis() - System.currentTimeMillis();
			final long period = 1000 * 60 * 60 * 24;

			// Broker的统计功能===1天统计一次
			this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
				@Override
				public void run() {
					try {
						BrokerController.this.getBrokerStats().record();
					} catch (Exception e) {
						log.error("schedule record error.", e);
					}
				}
			}, initialDelay, period, TimeUnit.MILLISECONDS);

			// 每隔5秒对ConsumerOffsetManager.offsetTable: ConcurrentHashMap<String/* topic@group */, ConcurrentHashMap<Integer, Long>>变量的内容进行持久化操作，
			// 持久化到consumerOffset.json文件中；
			this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
				@Override
				public void run() {
					try {
						BrokerController.this.consumerOffsetManager.persist();
					} catch (Exception e) {
						log.error("schedule persist consumerOffset error.", e);
					}
				}
			}, 1000 * 10, this.brokerConfig.getFlushConsumerOffsetInterval(), TimeUnit.MILLISECONDS);

			// 每小时扫描被删除的topic，将offsetTable中的对应的消费进度记录也删掉。大致逻辑是遍历ConsumerOffsetManager.offsetTable变量：
			this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
				@Override
				public void run() {
					try {
						BrokerController.this.consumerOffsetManager.scanUnsubscribedTopic();
					} catch (Exception e) {
						log.error("schedule scanUnsubscribedTopic error.", e);
					}
				}
			}, 10, 60, TimeUnit.MINUTES);

			// 检查Broker配置的Name Server地址是否为空
			if (this.brokerConfig.getNamesrvAddr() != null) {
				// 若不为空，则更新Broker的Netty客户端BrokerOuterAPI.NettyRemotingClient的Name Server地址
				this.brokerOuterAPI.updateNameServerAddressList(this.brokerConfig.getNamesrvAddr());
			} else if (this.brokerConfig.isFetchNamesrvAddrByAddressServer()) {
				// 若为空且配置了允许从地址服务器找Name Server地址，则启动定时任务
				this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

					@Override
					public void run() {
						try {
							BrokerController.this.brokerOuterAPI.fetchNameServerAddr();
						} catch (Exception e) {
							log.error("ScheduledTask fetchNameServerAddr exception", e);
						}
					}
				}, 1000 * 10, 1000 * 60 * 2, TimeUnit.MILLISECONDS);
			}

			if (BrokerRole.SLAVE == this.messageStoreConfig.getBrokerRole()) {
				// ======================若该Broker为备用========================
				// 在该Broker向NameServer注册的时候根据NameServer返回的主用Broker地址来设置HAClient.masterAddress值；
				// 说明在配置参数haMasterAddress设置的主用地址的优先级要高于NameServer返回的主用Broker地址。

				// 首先检查是否配置了主用Broker的地址，在备用Broker可以在broker.properties配置文件中设置haMasterAddress参数来指定主用Broker的地址
				if (this.messageStoreConfig.getHaMasterAddress() != null && this.messageStoreConfig.getHaMasterAddress().length() >= 6) {
					// 若设置了，则更新HAService.HAClient.masterAddress的值
					this.messageStore.updateHaMasterAddress(this.messageStoreConfig.getHaMasterAddress());
					this.updateMasterHAServerAddrPeriodically = false;
				} else {
					this.updateMasterHAServerAddrPeriodically = true;
				}

				// ======================若该Broker为备用========================
				/**
				 * 在集群模式下，为了保证高可用，必须要保证备用Broker与主用Broker信息是一致的，
				 * 设置同步Config文件的定时任务，每个60秒调用SlaveSynchronize.syncAll()方法发起向主用Broker进行一次config类文件的同步
				 */
				this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

					@Override
					public void run() {
						try {
							BrokerController.this.slaveSynchronize.syncAll();
						} catch (Exception e) {
							log.error("ScheduledTask syncAll slave exception", e);
						}
					}
				}, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
			} else {
				// =====================若该Broker为主用=========================
				// 则启动定时任务打印主用Broker和备用Broker在Commitlog上的写入位置相差多少个字节。
				this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

					@Override
					public void run() {
						try {
							BrokerController.this.printMasterAndSlaveDiff();
						} catch (Exception e) {
							log.error("schedule printMasterAndSlaveDiff error.", e);
						}
					}
				}, 1000 * 10, 1000 * 60, TimeUnit.MILLISECONDS);
			}
		}

		return result;
	}

	public void registerProcessor() {
		/**
		 * SendMessageProcessor
		 * 发送消息事件处理器
		 */
		SendMessageProcessor sendProcessor = new SendMessageProcessor(this);
		sendProcessor.registerSendMessageHook(sendMessageHookList);
		this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE, sendProcessor, this.sendMessageExecutor);
		this.remotingServer.registerProcessor(RequestCode.SEND_MESSAGE_V2, sendProcessor, this.sendMessageExecutor);
		// 将Consumer消费失败的消息写入延迟消息队列中
		this.remotingServer.registerProcessor(RequestCode.CONSUMER_SEND_MSG_BACK, sendProcessor, this.sendMessageExecutor);

		/**
		 * PullMessageProcessor
		 * 拉取消息事件处理器
		 */
		this.remotingServer.registerProcessor(RequestCode.PULL_MESSAGE, this.pullMessageProcessor, this.pullMessageExecutor);
		this.pullMessageProcessor.registerConsumeMessageHook(consumeMessageHookList);

		/**
		 * QueryMessageProcessor
		 * 查询消息事件处理器（包括客户端的心跳事件、注销事件、获取消费者列表事件、更新更新和查询消费进度consumerOffset）
		 */
		NettyRequestProcessor queryProcessor = new QueryMessageProcessor(this);
		this.remotingServer.registerProcessor(RequestCode.QUERY_MESSAGE, queryProcessor, this.pullMessageExecutor);
		this.remotingServer.registerProcessor(RequestCode.VIEW_MESSAGE_BY_ID, queryProcessor, this.pullMessageExecutor);

		/**
		 * ClientManageProcessor
		 * 客户端管理事件处理器
		 */
		ClientManageProcessor clientProcessor = new ClientManageProcessor(this);
		clientProcessor.registerConsumeMessageHook(this.consumeMessageHookList);
		this.remotingServer.registerProcessor(RequestCode.HEART_BEAT, clientProcessor, this.clientManageExecutor);
		this.remotingServer.registerProcessor(RequestCode.UNREGISTER_CLIENT, clientProcessor, this.clientManageExecutor);
		// 获取consumerGroup名下的所有Consumer的ClientId
		this.remotingServer.registerProcessor(RequestCode.GET_CONSUMER_LIST_BY_GROUP, clientProcessor, this.clientManageExecutor);

		/**
		 * Offset存储更新转移到ClientProcessor处理
		 */
		this.remotingServer.registerProcessor(RequestCode.UPDATE_CONSUMER_OFFSET, clientProcessor, this.clientManageExecutor);
		this.remotingServer.registerProcessor(RequestCode.QUERY_CONSUMER_OFFSET, clientProcessor, this.clientManageExecutor);

		/**
		 * EndTransactionProcessor
		 * 结束事务处理器
		 */
		// 处理结束事务消息的请求
		this.remotingServer.registerProcessor(RequestCode.END_TRANSACTION, new EndTransactionProcessor(this), this.sendMessageExecutor);

		/**
		 * Default
		 * 默认事件处理器
		 * 
		 // 更新创建Topic
		 RequestCode.UPDATE_AND_CREATE_TOPIC:
		 
		 // 删除Topic
		 RequestCode.DELETE_TOPIC_IN_BROKER:
		 
		 // 获取Topic配置
		 RequestCode.GET_ALL_TOPIC_CONFIG:
		
		 // 更新Broker配置 TODO 可能存在并发问题
		 RequestCode.UPDATE_BROKER_CONFIG:
		 
		 // 获取Broker配置
		 RequestCode.GET_BROKER_CONFIG:
		
		 // 根据时间查询Offset
		 RequestCode.SEARCH_OFFSET_BY_TIMESTAMP:
		 
		 RequestCode.GET_MAX_OFFSET:
		 RequestCode.GET_MIN_OFFSET:
		 RequestCode.GET_EARLIEST_MSG_STORETIME:
		
		 // 获取Broker运行时信息
		 RequestCode.GET_BROKER_RUNTIME_INFO:
		
		 // 锁队列与解锁队列
		 //=====================客户端顺序消费时锁住MessageQueue队列的请求==============
		 RequestCode.LOCK_BATCH_MQ:
		 
		 RequestCode.UNLOCK_BATCH_MQ:
		
		 // 订阅组配置
		 RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP:
		 
		 RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG:
		 RequestCode.DELETE_SUBSCRIPTIONGROUP:
		
		 // 统计信息，获取Topic统计信息
		 RequestCode.GET_TOPIC_STATS_INFO:
		 
		 // Consumer连接管理
		 RequestCode.GET_CONSUMER_CONNECTION_LIST:
		 
		 // Producer连接管理
		 RequestCode.GET_PRODUCER_CONNECTION_LIST:
		 
		 // 查询消费进度，订阅组下的所有Topic
		 RequestCode.GET_CONSUME_STATS:
		 
		 RequestCode.GET_ALL_CONSUMER_OFFSET:
		
		 // 定时进度
		 RequestCode.GET_ALL_DELAY_OFFSET:
		
		 // 调用客户端重置 offset
		 RequestCode.INVOKE_BROKER_TO_RESET_OFFSET:
		
		 // 调用客户端订阅消息处理
		 RequestCode.INVOKE_BROKER_TO_GET_CONSUMER_STATUS:
		
		 // 查询Topic被哪些消费者消费
		 RequestCode.QUERY_TOPIC_CONSUME_BY_WHO:
		
		 RequestCode.REGISTER_FILTER_SERVER:
		 
		 // 根据 topic 和 group 获取消息的时间跨度
		 RequestCode.QUERY_CONSUME_TIME_SPAN:
		 
		 RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_BROKER:
		
		 // 删除失效队列
		 RequestCode.CLEAN_EXPIRED_CONSUMEQUEUE:
		
		 RequestCode.GET_CONSUMER_RUNNING_INFO:
		
		 // 查找被修正 offset (转发组件）
		 RequestCode.QUERY_CORRECTION_OFFSET:
		
		 RequestCode.CONSUME_MESSAGE_DIRECTLY:
		 RequestCode.CLONE_GROUP_OFFSET:
		
		 // 查看Broker统计信息
		 RequestCode.VIEW_BROKER_STATS_DATA:
		 */
		AdminBrokerProcessor adminProcessor = new AdminBrokerProcessor(this);
		this.remotingServer.registerDefaultProcessor(adminProcessor, this.adminBrokerExecutor);
	}

	public Broker2Client getBroker2Client() {
		return broker2Client;
	}

	public BrokerConfig getBrokerConfig() {
		return brokerConfig;
	}

	public String getConfigDataVersion() {
		return this.configDataVersion.toJson();
	}

	public ConsumerManager getConsumerManager() {
		return consumerManager;
	}

	public ConsumerOffsetManager getConsumerOffsetManager() {
		return consumerOffsetManager;
	}

	public MessageStore getMessageStore() {
		return messageStore;
	}

	public void setMessageStore(MessageStore messageStore) {
		this.messageStore = messageStore;
	}

	public MessageStoreConfig getMessageStoreConfig() {
		return messageStoreConfig;
	}

	public NettyServerConfig getNettyServerConfig() {
		return nettyServerConfig;
	}

	public ProducerManager getProducerManager() {
		return producerManager;
	}

	public PullMessageProcessor getPullMessageProcessor() {
		return pullMessageProcessor;
	}

	public PullRequestHoldService getPullRequestHoldService() {
		return pullRequestHoldService;
	}

	public RemotingServer getRemotingServer() {
		return remotingServer;
	}

	public void setRemotingServer(RemotingServer remotingServer) {
		this.remotingServer = remotingServer;
	}

	public SubscriptionGroupManager getSubscriptionGroupManager() {
		return subscriptionGroupManager;
	}

	public void shutdown() {
		if (this.brokerStatsManager != null) {
			this.brokerStatsManager.shutdown();
		}

		if (this.clientHousekeepingService != null) {
			this.clientHousekeepingService.shutdown();
		}

		if (this.pullRequestHoldService != null) {
			this.pullRequestHoldService.shutdown();
		}

		if (this.remotingServer != null) {
			this.remotingServer.shutdown();
		}

		if (this.messageStore != null) {
			this.messageStore.shutdown();
		}

		this.scheduledExecutorService.shutdown();
		try {
			this.scheduledExecutorService.awaitTermination(5000, TimeUnit.MILLISECONDS);
		} catch (InterruptedException e) {
		}

		this.unregisterBrokerAll();

		if (this.sendMessageExecutor != null) {
			this.sendMessageExecutor.shutdown();
		}

		if (this.pullMessageExecutor != null) {
			this.pullMessageExecutor.shutdown();
		}

		if (this.adminBrokerExecutor != null) {
			this.adminBrokerExecutor.shutdown();
		}

		if (this.brokerOuterAPI != null) {
			this.brokerOuterAPI.shutdown();
		}

		this.consumerOffsetManager.persist();

		if (this.filterServerManager != null) {
			this.filterServerManager.shutdown();
		}
	}

	private void unregisterBrokerAll() {
		this.brokerOuterAPI.unregisterBrokerAll(//
				this.brokerConfig.getBrokerClusterName(), //
				this.getBrokerAddr(), //
				this.brokerConfig.getBrokerName(), //
				this.brokerConfig.getBrokerId());
	}

	public String getBrokerAddr() {
		String addr = this.brokerConfig.getBrokerIP1() + ":" + this.nettyServerConfig.getListenPort();
		return addr;
	}

	public void start() throws Exception {
		// 启动服务
		// flushConsumeQueueService====逻辑队列刷盘服务，每隔1秒钟就将ConsumeQueue逻辑队列、TransactionStateService.TranRedoLog变量的数据持久化到磁盘物理文件中
		// flushCommitLogService====逻辑队列刷盘服务
		// storeStatsService===存储层内部统计服务
		// scheduleMessageService==在Broker是主用模式的时候，启动ScheduleMessageService线程服务；在启动的过程中，向每个级别的对应的队列都增加定时任务，周期性的检查队列中是否有延迟消息存在，若有则到期后就执行该延迟消息，执行的目的是将真正的消息写入commitlog中，并生成consumequeue和index数据。
		// reputMessageService====会一直不间断的监听reputFromOffset偏移量之后的commitlog数据，若commitlog数据有增加，即reputFromOffset偏移量之后新增了数据，则获取出来并创建consumequeue和index，同时更新reputFromOffset偏移量
		// haService===用于commitlog数据的主备同步
		if (this.messageStore != null) {
			this.messageStore.start();
		}

		// 启动Broker的Netty服务端NettyRemotingServer。监听消费者或生产者发起的请求信息并处理；
		if (this.remotingServer != null) {
			this.remotingServer.start();
		}

		/**
		 * 启动BrokerOuterAPI中的NettyRemotingClient，
		 * 即建立与NameServer的链接，用于自身Broker与其他模块的RPC功能调用；
		 * 包括获取NameServer的地址、注册Broker、注销Broker、获取Topic配置、获取消息进度信息、获取订阅关系等RPC功能。
		 */
		if (this.brokerOuterAPI != null) {
			this.brokerOuterAPI.start();
		}

		/**
		 * ===============================对未拉取到的消息进行重试==================================
		 * 主要是对于未拉取到的信息进行补偿，在后台每个1秒钟进行尝试拉取信息
		 * 
		 * 启动拉消息管理服务PullRequestHoldService，
		 * 当拉取消息时未发现消息，则初始化PullRequeset对象放入该服务线程的pullRequestTable列表中，
		 * 由PullRequestHoldService每隔1秒钟就检查一遍每个PullRequeset对象要读取的数据位置在consumequeue中是否已经有数据了，若有则交由PullManageProcessor处理。
		 */
		if (this.pullRequestHoldService != null) {
			this.pullRequestHoldService.start();
		}

		/**
		 * 启动ClientHousekeepingService服务，在启动过程中设置定时任务，该定时任务每隔10秒就检查一次客户端的链接情况，清除不活动的链接（即在120秒以内没有数据交互了），
		 * 包括：Producer的链接、Consumer的链接、过滤服务器的链接
		 */
		if (this.clientHousekeepingService != null) {
			this.clientHousekeepingService.start();
		}

		/**
		 * 启动FilterServerManager，每隔30秒定期一次检查Filter Server个数，若没有达到16个则创建；
		 */
		if (this.filterServerManager != null) {
			this.filterServerManager.start();
		}

		// 立即向NameServer注册Broker
		this.registerBrokerAll(true, false);

		// 此后每隔30秒调用一次该方法向NameServer注册；
		this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

			@Override
			public void run() {
				try {
					BrokerController.this.registerBrokerAll(true, false);
				} catch (Exception e) {
					log.error("registerBrokerAll Exception", e);
				}
			}
		}, 1000 * 10, 1000 * 30, TimeUnit.MILLISECONDS);

		if (this.brokerStatsManager != null) {
			this.brokerStatsManager.start();
		}

		/**
		 * 启动每隔5秒对未使用的topic数据进行清理的定时任务。该定时任务调用DefaultMessageStore.cleanUnusedTopic(Set<String>topics)方法。
		 */
		this.addDeleteTopicTask();
	}

	/**
	 * 立即向NameServer注册Broker，即对NameServer中的topic信息更新。
	 * @param checkOrderConfig
	 * @param oneway
	 */
	public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway) {
		// 1、将TopicConfigManager.topicConfigTable变量序列化成TopicConfigSerializeWrapper对象；
		TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();

		if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission()) || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {
			ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<String, TopicConfig>(topicConfigWrapper.getTopicConfigTable());
			for (TopicConfig topicConfig : topicConfigTable.values()) {
				topicConfig.setPerm(this.getBrokerConfig().getBrokerPermission());
			}
			topicConfigWrapper.setTopicConfigTable(topicConfigTable);
		}

		// 向NameServer注册Broker，即对NameServer中的topic信息更新。
		RegisterBrokerResult registerBrokerResult = this.brokerOuterAPI.registerBrokerAll(//
				this.brokerConfig.getBrokerClusterName(), //
				this.getBrokerAddr(), //
				this.brokerConfig.getBrokerName(), //
				this.brokerConfig.getBrokerId(), //
				this.getHAServerAddr(), // 该Broker的本地IP地址:端口号（默认为10912）
				topicConfigWrapper, //
				this.filterServerManager.buildNewFilterServerList(), //
				oneway);

		if (registerBrokerResult != null) {
			/**
			 * 判断updateMasterHAServerAddrPeriodically标注位以及NameServer返回的HaServerAddr地址是否为空
			 */
			if (this.updateMasterHAServerAddrPeriodically /* 299行在初始化时若Broker的配置文件中没有haMasterAddress参数配置，则标记为true，表示注册之后需要更新主用Broker地址 */ && registerBrokerResult.getHaServerAddr() != null) {
				/**
				 * 若标记位是true且返回的HaServerAddr不为空，则用HaServerAddr地址更新HAService.HAClient.masterAddress的值；该HAClient.masterAddress值用于主备Broker之间的commitlog数据同步之用；
				 */
				this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());
			}

			/**
			 * 用NameServer返回的MasterAddr值更新SlaveSynchronize.masterAddr值，用于主备Broker同步Config文件使用；
			 */
			this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());

			if (checkOrderConfig) {
				/**
				 * 根据Name Server上的配置参数ORDER_TOPIC_CONFIG的值来更新Broker端的TopicConfigManager.topicConfigTable变量值，对于在ORDER_TOPIC_CONFIG中配置了的topic，将该topic的配置参数order置为true，否则将该topic的配置参数order置为false；
				 */
				this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());
			}
		}
	}

	public TopicConfigManager getTopicConfigManager() {
		return topicConfigManager;
	}

	public void setTopicConfigManager(TopicConfigManager topicConfigManager) {
		this.topicConfigManager = topicConfigManager;
	}

	public String getHAServerAddr() {
		String addr = this.brokerConfig.getBrokerIP2() + ":" + this.messageStoreConfig.getHaListenPort();
		return addr;
	}

	public void updateAllConfig(Properties properties) {
		MixAll.properties2Object(properties, brokerConfig);
		MixAll.properties2Object(properties, nettyServerConfig);
		MixAll.properties2Object(properties, nettyClientConfig);
		MixAll.properties2Object(properties, messageStoreConfig);
		this.configDataVersion.nextVersion();
		this.flushAllConfig();
	}

	private void flushAllConfig() {
		String allConfig = this.encodeAllConfig();
		try {
			MixAll.string2File(allConfig, BrokerPathConfigHelper.getBrokerConfigPath());
			log.info("flush broker config, {} OK", BrokerPathConfigHelper.getBrokerConfigPath());
		} catch (IOException e) {
			log.info("flush broker config Exception, " + BrokerPathConfigHelper.getBrokerConfigPath(), e);
		}
	}

	public String encodeAllConfig() {
		StringBuilder sb = new StringBuilder();
		{
			Properties properties = MixAll.object2Properties(this.brokerConfig);
			if (properties != null) {
				sb.append(MixAll.properties2String(properties));
			} else {
				log.error("encodeAllConfig object2Properties error");
			}
		}

		{
			Properties properties = MixAll.object2Properties(this.messageStoreConfig);
			if (properties != null) {
				sb.append(MixAll.properties2String(properties));
			} else {
				log.error("encodeAllConfig object2Properties error");
			}
		}

		{
			Properties properties = MixAll.object2Properties(this.nettyServerConfig);
			if (properties != null) {
				sb.append(MixAll.properties2String(properties));
			} else {
				log.error("encodeAllConfig object2Properties error");
			}
		}

		{
			Properties properties = MixAll.object2Properties(this.nettyClientConfig);
			if (properties != null) {
				sb.append(MixAll.properties2String(properties));
			} else {
				log.error("encodeAllConfig object2Properties error");
			}
		}
		return sb.toString();
	}

	public RebalanceLockManager getRebalanceLockManager() {
		return rebalanceLockManager;
	}

	public SlaveSynchronize getSlaveSynchronize() {
		return slaveSynchronize;
	}

	public BrokerOuterAPI getBrokerOuterAPI() {
		return brokerOuterAPI;
	}

	public ExecutorService getPullMessageExecutor() {
		return pullMessageExecutor;
	}

	public void setPullMessageExecutor(ExecutorService pullMessageExecutor) {
		this.pullMessageExecutor = pullMessageExecutor;
	}

	public BrokerStats getBrokerStats() {
		return brokerStats;
	}

	public void setBrokerStats(BrokerStats brokerStats) {
		this.brokerStats = brokerStats;
	}

	public BlockingQueue<Runnable> getSendThreadPoolQueue() {
		return sendThreadPoolQueue;
	}

	public FilterServerManager getFilterServerManager() {
		return filterServerManager;
	}

	public BrokerStatsManager getBrokerStatsManager() {
		return brokerStatsManager;
	}

	private void printMasterAndSlaveDiff() {
		long diff = this.messageStore.slaveFallBehindMuch();

		// XXX: warn and notify me
		log.info("slave fall behind master, how much, {} bytes", diff);
	}

	public void addDeleteTopicTask() {
		this.scheduledExecutorService.schedule(new Runnable() {
			@Override
			public void run() {
				// 调用DefaultMessageStore.cleanUnusedTopic(Set<String>topics)方法对未使用的topic数据进行清理
				int removedTopicCnt = BrokerController.this.messageStore.cleanUnusedTopic(BrokerController.this.getTopicConfigManager().getTopicConfigTable().keySet()/* 请求参数topics集合取至TopicConfigManager.topicConfigTable列表的key值集合；由Broker的心跳检测功能来维护该topicConfigTable列表的数据，该列表数据对应的是topics.json文件中的数据； */);
				log.info("addDeleteTopicTask removed topic count {}", removedTopicCnt);
			}
		}, 5, TimeUnit.MINUTES);
	}

	private final List<SendMessageHook> sendMessageHookList = new ArrayList<SendMessageHook>();

	public void registerSendMessageHook(final SendMessageHook hook) {
		this.sendMessageHookList.add(hook);
		log.info("register SendMessageHook Hook, {}", hook.hookName());
	}

	private final List<ConsumeMessageHook> consumeMessageHookList = new ArrayList<ConsumeMessageHook>();

	public void registerConsumeMessageHook(final ConsumeMessageHook hook) {
		this.consumeMessageHookList.add(hook);
		log.info("register ConsumeMessageHook Hook, {}", hook.hookName());
	}

	public void registerServerRPCHook(RPCHook rpcHook) {
		getRemotingServer().registerRPCHook(rpcHook);
	}

	public void registerClientRPCHook(RPCHook rpcHook) {
		this.getBrokerOuterAPI().registerRPCHook(rpcHook);
	}

	public InetSocketAddress getStoreHost() {
		return storeHost;
	}

	public void setStoreHost(InetSocketAddress storeHost) {
		this.storeHost = storeHost;
	}
}
